package com.bw.flinkstreaming.state2.job5;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 * 需求：实现相同key的value的拼接
 * 101，50
 * 101，60
 *
 * 101，50-60-70....
 * */
public class AggregatingStateExe extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,String>> {

    //定义 aggregatingState
    private AggregatingState<Long,String> aggregatingState;

    /**
     * 初始化state
     * */
    @Override
    public void open(Configuration parameters) throws Exception {
        AggregatingStateDescriptor aggregatingStateDescriptor = new AggregatingStateDescriptor<Long,String,String>("aggregatingState",
                /**
                 *  * @param <IN>  The type of the values that are aggregated (input values)
                 *  * @param <ACC> The type of the accumulator (intermediate aggregate state).
                 *  * @param <OUT> The type of the aggregated result
                 * */
                new AggregateFunction<Long, String, String>() {

                    /**
                     * start
                     * */
                    @Override
                    public String createAccumulator() {
                        return "start";
                    }

                    /**
                     * 当前节点处理的逻辑
                     * */
                    @Override
                    public String add(Long value, String accumulator) {
                        //第一次处理  start50
                        if ("start".equals(accumulator)) {
                            return accumulator + value;
                        }
                        //不是第一次 start50-60
                        return accumulator + "-" + value;
                    }

                    @Override
                    public String getResult(String accumulator) {
                        return accumulator;
                    }

                    /**
                     * 分区之间数据合并的时候处理的方法
                     * */
                    @Override
                    public String merge(String a, String b) {
                        return a + "-" + b;
                    }
                },String.class);
        aggregatingState = getRuntimeContext().getAggregatingState(aggregatingStateDescriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, String>> out) throws Exception {
        aggregatingState.add(value.f1);
        out.collect(Tuple2.of(value.f0,aggregatingState.get()));
    }
}
